package top.wintp.offlinedataanalysis.anlyser.mr.nu;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

import top.wintp.offlinedataanalysis.anlyser.dim.StatsUserDimension;
import top.wintp.offlinedataanalysis.anlyser.value.map.TimeOutputValue;
import top.wintp.offlinedataanalysis.anlyser.value.reduce.MapWritableValue;
import top.wintp.offlinedataanalysis.common.KpiType;

/**
 * @author: pyfysf
 * <p>
 * @qq: 337081267
 * <p>
 * @CSDN: http://blog.csdn.net/pyfysf
 * <p>
 * @blog: http://wintp.top
 * <p>
 * @email: pyfysf@163.com
 * <p>
 * @time: 2019/9/5
 */
public class NewInstallUserReducer extends Reducer<StatsUserDimension, TimeOutputValue, StatsUserDimension, MapWritableValue> {
    private static final Logger logger = LoggerFactory.getLogger(NewInstallUserReducer.class);

    /**
     * 初始化输出的value
     */
    private MapWritableValue mapWritableValue = new MapWritableValue();
    /**
     * 输出的value -map的val
     */
    private IntWritable mapWritableVal = new IntWritable();
    /**
     * 去重的集合
     */
    private Set<String> uuidSet = new HashSet<>();
    /**
     * 输出的value中的数据封装map
     */
    private MapWritable mapWritable = new MapWritable();
    /**
     * 输出value的key
     */
    private IntWritable mapWritableKey = new IntWritable(-1);

    @Override
    protected void reduce(StatsUserDimension key, @org.jetbrains.annotations.NotNull Iterable<TimeOutputValue> values, Context context) throws IOException, InterruptedException {

        logger.info("NewInstallUserReducer  reduce()   REDUCE_KPI   " + key.getStatsCommon().getKpi().getKpiName());

        //    清空集合
        uuidSet.clear();
        //    遍历value值
        for (TimeOutputValue value : values) {
            uuidSet.add(value.getId());
        }

        //    组合key-value输出
        mapWritableVal.set(uuidSet.size());

        mapWritable.put(mapWritableKey, mapWritableVal);

        mapWritableValue.setValue(mapWritable);

        mapWritableValue.setKpi(KpiType.valueOfName(key.getStatsCommon().getKpi().getKpiName()));

        //    数据输出
        context.write(key, mapWritableValue);
    }
}
